package com.zgkxzx.android.rx;

import com.zgkxzx.android.rx.schedule.Scheduler;

import java.util.ArrayList;
import java.util.List;

/**
 * 创建者   zgkxzx
 * 创建日期 2017/12/8.
 * 功能描述 被观察者
 */

public class Observable<T> {
    private OnSubscribe<T> onSubscribe;

    public Observable(OnSubscribe<T> onSubscribe) {
        this.onSubscribe = onSubscribe;
    }

    public static <T> Observable<T> create(OnSubscribe<T> onSubscribe) {
        return new Observable(onSubscribe);
    }

    //订阅
    public void subscribe(Subscriber<T> subscriber) {
        subscriber.onStart();
        onSubscribe.call(subscriber);
    }


    //from
    public static <T> Observable from(final T[] ts) {
        return create(new OnSubscribe<T>() {
            @Override
            public void call(Subscriber<T> subscriber) {
                for (T t : ts)
                    subscriber.onNext(t);
                subscriber.onCompleted();
            }
        });
    }

    //just方法
    public static <T> Observable just(T... ts) {
        return from(ts);
    }


    //map方法
    public <R> Observable<R> map(final Func1<T, R> func1) {
        return create(new OnSubscribe<R>() {
            @Override
            public void call(final Subscriber<R> subscriber) {
                subscribe(new Subscriber<T>() {
                    @Override
                    public void onError(Throwable e) {
                        subscriber.onError(e);
                    }

                    @Override
                    public void onCompleted() {
                        subscriber.onCompleted();
                    }

                    @Override
                    public void onNext(T t) {
                        subscriber.onNext(func1.call(t));
                    }
                });
            }
        });
    }



    //flatMap
    public <R> Observable flatMap(final Func1<T, R> fun1) {
        return create(new OnSubscribe() {
            @Override
            public void call(final Subscriber subscriber) {
                subscribe(new Subscriber<T>() {
                    List<R> rs = new ArrayList<R>();

                    @Override
                    public void onError(Throwable e) {
                        subscriber.onError(e);
                    }

                    @Override
                    public void onCompleted() {
                        for (R r : rs)
                            subscriber.onNext(r);
                        subscriber.onCompleted();
                        //Log.d("dd", "-->onCompleted");
                    }

                    @Override
                    public void onNext(T t) {
                        rs.add(fun1.call(t));
                        //Log.d("dd", "-->" + t);
                    }
                });


            }
        });

    }

    ////////////////////////基本创建变换//////////////////////
    private Observable baseCreate(final IBaseSubscriber<T> baseSubscriber) {
        return create(new OnSubscribe() {
            @Override
            public void call(Subscriber subscriber) {
                baseSubscriber.newSubscriber(subscriber);
            }
        });
    }

    public interface IBaseSubscriber<T> {
        Subscriber<T> newSubscriber(Subscriber subscriber);
    }
    //////////////////////////////////////////////////////////////////////


    /*public Observable test(final Func1<T, Boolean> fun1) {

        return baseTransfer(new IBaseSubscriber<T>() {
            @Override
            public Subscriber<T> newSubscriber(Subscriber subscriber) {
                return new Subscriber<T>() {
                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onCompleted() {

                    }

                    @Override
                    public void onNext(T t) {

                    }
                };
            }
        });
    }*/

    //filter
    public Observable filter(final Func1<T, Boolean> fun1) {
        return create(new OnSubscribe() {
            @Override
            public void call(final Subscriber subscriber) {
                subscribe(new Subscriber<T>() {
                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onCompleted() {
                        subscriber.onCompleted();
                    }

                    @Override
                    public void onNext(T t) {
                        if (fun1.call(t))
                            subscriber.onNext(t);
                    }
                });
            }
        });
    }

//    .subscribeOn(Schedulers.io()) // 指定 subscribe() 发生在 IO 线程
//    .observeOn(AndroidSchedulers.mainThread()) // 指定 Subscriber 的回调发生在主线程

    //subscriber的线程控制
    public Observable subscribeOn(final Scheduler scheduler) {
        return create(new OnSubscribe() {
            @Override
            public void call(final Subscriber subscriber) {
                subscriber.onStart();
                scheduler.createWorker().schedule(new Action0() {
                    @Override
                    public void call() {
                        onSubscribe.call(subscriber);
                    }
                });
            }
        });
    }

    public Observable observeOn(final Scheduler scheduler) {
        return create(new OnSubscribe() {
            @Override
            public void call(final Subscriber subscriber) {
                subscribe(new Subscriber<T>() {
                    @Override
                    public void onError(final Throwable e) {
                        scheduler.createWorker().schedule(new Action0() {
                            @Override
                            public void call() {
                                subscriber.onError(e);
                            }
                        });
                    }

                    @Override
                    public void onCompleted() {
                        scheduler.createWorker().schedule(new Action0() {
                            @Override
                            public void call() {
                                subscriber.onCompleted();
                            }
                        });
                    }

                    @Override
                    public void onNext(final T t) {
                        scheduler.createWorker().schedule(new Action0() {
                            @Override
                            public void call() {
                                subscriber.onNext(t);
                            }
                        });
                    }
                });
            }
        });
    }


}
